package com.chj.Sender;

import com.rabbitmq.client.*;
import org.springframework.amqp.core.TopicExchange;

import java.io.IOException;

/**
 * Created by Administrator on 2019/3/14.
 */
public class ConfirmSender1{
    private final static String QUEUE_NAME = "confirm";

    public static void main(String[] args) throws Exception{
        /**
         * 创建连接连接到RabbitMQ
         */
        ConnectionFactory factory = new ConnectionFactory();
        // 设置RabbitMQ所在主机ip或者主机名
        factory.setUsername("admin");
        factory.setPassword("admin");
        factory.setHost("127.0.0.1");
        factory.setVirtualHost("/");
        factory.setPort(5672);
        // 创建一个连接
        Connection connection = factory.newConnection();
        // 创建一个信道
        Channel channel = connection.createChannel();
        // 指定一个队列  String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 发送的消息
        String message = "This is a confirm message！";
        channel.confirmSelect();

        final long start = System.currentTimeMillis();
        //发送持久化消息
//        for (int i = 0; i < 5; i++) {
//            //第一个参数是exchangeName(默认情况下代理服务器端是存在一个""名字的exchange的,
//            //因此如果不创建exchange的话我们可以直接将该参数设置成"",如果创建了exchange的话
//            //我们需要将该参数设置成创建的exchange的名字),第二个参数是路由键
////            TopicExchange conformExchenge = new TopicExchange("conformExchenge");
//            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, (" Confirm模式， 第" + (i + 1) + "条消息").getBytes());
//            if (channel.waitForConfirms()) {
//                System.out.println("发送成功");
//            }else{
//                // 进行消息重发
//            }
//        }

        // 批量Confirm模式
        for(int i=0;i<5;i++){
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_BASIC, (" Confirm模式， 第" + (i + 1) + "条消息").getBytes());
            // 对每条消息进行监听处理
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("nack: deliveryTag = " + deliveryTag + " multiple: " + multiple);
                }
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("ack: deliveryTag = " + deliveryTag + " multiple: " + multiple);
                }
            });
        }
        if(channel.waitForConfirms()){
            System.out.println("发送成功");
        }else{
            // 进行消息重发
        }


        System.out.println("执行waitForConfirms耗费时间: " + (System.currentTimeMillis() - start) + "ms");
        // 关闭频道和连接
        channel.close();
        connection.close();
    }


}
